package net.wicp.tams.common.others.kafka;

import java.lang.reflect.ParameterizedType;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.Properties;

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;

import lombok.extern.slf4j.Slf4j;
import net.wicp.tams.common.Result;

@Slf4j
public abstract class KafkaConsumerGroup<T> {

	private List<KafkaConsumerGroupThread> consumerThreadList = new ArrayList<KafkaConsumerGroupThread>();

	private String groupId;
	private String topic;
	private IConsumer<T> doConsumer;

	public KafkaConsumerGroup(String groupId, String topic, IConsumer<T> doConsumer) {
		this.groupId = groupId;
		this.topic = topic;
		this.doConsumer = doConsumer;
		for (int i = 0; i < 3; i++) {
			KafkaConsumerGroupThread consumerThread = new KafkaConsumerGroupThread();
			consumerThreadList.add(consumerThread);
		}
	}

	public Class<T> getTClass() {
		Class<T> tClass = (Class<T>) ((ParameterizedType) this.getClass().getGenericSuperclass())
				.getActualTypeArguments()[0];
		 //System.out.println(((ParameterizedType)new KafkaConsumerGroup().getClass().getGenericSuperclass()).getActualTypeArguments()[0]);  		 
		return tClass;
	}

	public void start() {
		for (KafkaConsumerGroupThread item : consumerThreadList) {
			Thread thread = new Thread(item);
			thread.start();
		}
	}

	private class KafkaConsumerGroupThread implements Runnable {
		private KafkaConsumer<String, T> kafkaConsumer;

		public KafkaConsumerGroupThread() {
			Properties props = KafkaTools.getProps(false);
			props.put("group.id", groupId);
			props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
			props.put("value.deserializer", KafkaTools.getValueProp(getTClass(), false));			
			
			/*Properties props = new Properties();
			 props.put("bootstrap.servers", "zk-kafka-01:9092,zk-kafka-02:9092,zk-kafka-03:9092,zk-kafka-04:9092");
			 props.put("group.id", groupId);
			 props.put("enable.auto.commit", "true");
			 props.put("auto.commit.interval.ms", "1000");
			 props.put("session.timeout.ms", "30000");
			 props.put("auto.offset.reset", "earliest");
			 props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
			 props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");*/
		        
			log.info("kafka consumer 参数:");
			for (Object propele : props.keySet()) {
				log.info("{}:{}", propele, props.get(propele));
			}
			this.kafkaConsumer = new KafkaConsumer<String, T>(props);
			this.kafkaConsumer.subscribe(Arrays.asList(topic));
		}

		@Override
		public void run() {
			while (true) {
				ConsumerRecords<String, T> consumerRecords = kafkaConsumer.poll(100);
				for (ConsumerRecord<String, T> item : consumerRecords) {
					Result doWithRecord = doConsumer.doWithRecord(item);
					log.info(doWithRecord.getMessage());
				}
			}
		}
	}
}
